#include "stdafx.h"
#include <boost/asio.hpp>
#include "session_imp.h"
#include <boost/bind.hpp>
#include "listener_imp.h"
#include <ut_debug.h>

#ifdef WIN32
#pragma comment(lib, "libeay32.lib")
#pragma comment(lib, "ssleay32.lib")
#endif

using namespace std;

namespace utility
{
namespace network
{
	static const int MAGIC_NUM = 0x98765432;
	struct MSG_HEADER
	{
		uint32 length;
		uint32 magic_num;
	};

	session_impl::session_impl(const boost::asio::io_service & io_service):
	socket_(const_cast<boost::asio::io_service &>(io_service))       
	{
        event_handler_ = NULL;
		listener_ = NULL;
        recv_buffer_used_ = 0;
        max_rcv_buf_size = 8192;
        recv_buffer_ = (char *)malloc(max_rcv_buf_size);
        set_disconnected_ = false;
    }
    
    session_impl::~session_impl()
    {
        free(recv_buffer_);
        set_disconnected_ = false;
    }
    
    session_impl::lowest_layer_type & session_impl::socket()
    {
        return socket_.lowest_layer();
    }


    void session_impl::disconnect(bool bForce)
    {
        set_disconnected_ = true;
        if (bForce || send_buffer_.empty())
        {
            _disconnect();
        }
    }

    void session_impl::set_event_handler(session_event_handler::pointer event_handler)
    {
        event_handler_ = event_handler;
    }

    session_event_handler::pointer session_impl::get_event_handler()
    {
        return event_handler_;
    }

    void session_impl::send(void *message, size_t message_length)
    {
        if (set_disconnected_)
            return;

        if (message_length == 0)
            return;

        boost::lock_guard<lock_type> guard(lock_);
        bool buffer_empty = send_buffer_.empty();
        string temp((char*)message, message_length);
        _encode(temp);
        send_buffer_.push(temp);
        if (buffer_empty)
            _send();
    }

    void session_impl::accept_handler(const boost::system::error_code &error)
    {
        try
        {
            if (!error)
            {
                remote_ = socket().remote_endpoint();
                local_ = socket().local_endpoint();
            }

            if (event_handler_)
            {
                if (!error)
                {
                    event_handler_->on_established();
                }
                else
                {
                    event_handler_->on_error(error.value(), error.message().c_str());
                }
            }
        }
        catch(std::exception & e)
        {
            LOG(CRITICAL) << "***********************" << __FUNCTION__ << e.what();
            disconnect();
        }
        catch(...)
        {
            disconnect();
        }

    }

    void session_impl::start(const boost::system::error_code& error, bool isServer)
    {
        accept_handler(error);
        _receive();
    }

    void session_impl::_receive()
    {
        socket_.async_receive(
            boost::asio::buffer(recv_buffer_ + recv_buffer_used_ ,max_rcv_buf_size - recv_buffer_used_),
            boost::bind(&session_impl::_receive_handler, shared_from_this(),
            boost::asio::placeholders::error,
            boost::asio::placeholders::bytes_transferred)
            );
    }

    void session_impl::_send()
    {
        socket_.async_send(send_buffer_,
            boost::bind(&session_impl::_send_handler, shared_from_this(), boost::asio::placeholders::error,
            boost::asio::placeholders::bytes_transferred));
    }

    void session_impl::_send_handler(const boost::system::error_code &error, size_t bytes_transferred)
    {
        if (!error)
        {
            boost::lock_guard<lock_type> guard(lock_);
            send_buffer_.consume(bytes_transferred);
            if (!send_buffer_.empty())
            {
                _send();
                return;
            }
        }
        else
        {
            if (event_handler_)
                event_handler_->on_error(error.value(), error.message().c_str());
        }

        if (set_disconnected_)
            _disconnect();
    }

    void session_impl::_decode(size_t bytes_transferred)
    {
        size_t decoded_buffer_size = 0;
        
        recv_buffer_used_ += bytes_transferred;
        //if (recv_buffer_used_ < sizeof(MSG_HEADER) )
        //    return;
        while(decoded_buffer_size + sizeof (MSG_HEADER) < recv_buffer_used_)
        {

            MSG_HEADER * msg_header = (MSG_HEADER *)(recv_buffer_ + decoded_buffer_size);
			uint32 item_length = ntohl(msg_header->length);
			uint32 magic_num = ntohl(msg_header->magic_num);
			if (magic_num != MAGIC_NUM)
			{
				disconnect();
				return;
			}
            uint32 message_length = item_length;// + sizeof(item_length);
            if (decoded_buffer_size + message_length <= recv_buffer_used_)
                event_handler_->on_receive(recv_buffer_ + decoded_buffer_size + sizeof(MSG_HEADER), item_length - sizeof(MSG_HEADER));
            else
                break;
            decoded_buffer_size += message_length;
        }

        if (decoded_buffer_size > 0)
        {
            if (decoded_buffer_size != recv_buffer_used_)
                memmove(recv_buffer_, recv_buffer_ + decoded_buffer_size, recv_buffer_used_ - decoded_buffer_size);
            recv_buffer_used_-= decoded_buffer_size;
        }

    }

    void session_impl::_encode(string & message)
    {
        uint32 length = message.size() + sizeof(MSG_HEADER);
		MSG_HEADER msg_header;
		msg_header.length = htonl(length);
		msg_header.magic_num = htonl(MAGIC_NUM);
        string header((char *)&msg_header, sizeof(msg_header));
        message = header + message;
    }

    void session_impl::_receive_handler(const boost::system::error_code &error, size_t bytes_transferred)
    {
        if ( !error )
        {
            if ( bytes_transferred > 0)
            {
                _decode(bytes_transferred);
                _receive();
                return;
            }
        }
        else if ( error == boost::asio::error::eof )
        {
            disconnect();
            if (event_handler_)
                event_handler_->on_disconnected();
            
        }
        else
        {
            disconnect();
            if (event_handler_)
                event_handler_->on_error(error.value(),error.message().c_str());
            
        }
    }

    uint32 session_impl::get_remote_ip()
    {
        try{
            return ntohl(remote_.address().to_v4().to_ulong());
        }
        catch(std::exception & e)
        {
            LOG(CRITICAL) << "***********************" << __FUNCTION__ << ':' <<e.what();
            return 0;
        }
    }

    std::string session_impl::get_remote_ip_as_string()
    {
        try
        {
            return remote_.address().to_string();
        }
        catch(std::exception & e)
        {
            LOG(CRITICAL) << "***********************" << __FUNCTION__ << ':' <<e.what();
            static string empty = "";
            return empty;
        }
        
    }

    uint16 session_impl::get_remote_port()
    {
        try{
            return remote_.port();
        }
        catch(std::exception & e)
        {
            LOG(CRITICAL) << "***********************" << __FUNCTION__ << ':' <<e.what();
            return 0;
        }
    }

    uint32 session_impl::get_local_ip()
    {
        try{
            return ntohl(local_.address().to_v4().to_ulong());
        }
        catch(std::exception & e)
        {
            LOG(CRITICAL) << "***********************" << __FUNCTION__ << ':' <<e.what();
            return 0;
        }
    }

    std::string session_impl::get_local_ip_as_string()
    {
        try
        {
            return local_.address().to_string();
        }
        catch(std::exception & e)
        {
            LOG(CRITICAL) << "***********************" << __FUNCTION__ << ':' <<e.what();
            static string empty = "";
            return empty;
        }
    }

    uint16 session_impl::get_local_port()
    {
        try
        {
            return local_.port();
        }
        catch(std::exception & e)
        {
            LOG(CRITICAL) << "***********************" << __FUNCTION__ << ':' <<e.what();
            return 0;
        }
        
    }

    void session_impl::set_listener(listener_imp * l)
    {
        listener_ = l;
    }

    void session_impl::stop()
    {
        try
        {
            socket().close();
        }
        catch(std::exception & e)
        {
            LOG(CRITICAL) << "***********************" << __FUNCTION__ << ':' <<e.what();
        }
    }

    void session_impl::reset_rcv_buf_size( uint32 nSize )
    {
        recv_buffer_ = (char *)realloc(recv_buffer_, nSize);
        max_rcv_buf_size = nSize;
    }

    void session_impl::_disconnect()
    {
        if (listener_)
            listener_->stop_session(shared_from_this());
        else
            stop();
        set_disconnected_ = false;
    }

    /************************************************************************/
    /* Implementation of secure_session_impl                                */
    /************************************************************************/
    secure_session_impl::secure_session_impl(const boost::asio::io_service& io_service, const boost::asio::ssl::context& context):
        session_impl(io_service),
        ssl_socket_(const_cast<boost::asio::io_service&>(io_service), 
                    const_cast<boost::asio::ssl::context&>(context))
    {

    }
    secure_session_impl::lowest_layer_type & secure_session_impl::socket()
    {
        return ssl_socket_.lowest_layer();
    }

    void secure_session_impl::start(const boost::system::error_code& error, bool isServer)
    {
        if (isServer)
            ssl_socket_.async_handshake(boost::asio::ssl::stream_base::server,
                                        boost::bind(&secure_session_impl::_handshake_handler, this,
                                                    boost::asio::placeholders::error));
        else 
            ssl_socket_.async_handshake(boost::asio::ssl::stream_base::client,
                                        boost::bind(&secure_session_impl::_handshake_handler, this,
                                                    boost::asio::placeholders::error));

    }

    void secure_session_impl::_handshake_handler(const boost::system::error_code& error)
    {
        if (!error)
        {
            _receive();
        }
        else
        {
            disconnect();
            //if (event_handler_)
            //    event_handler_->on_error(error.value(),error.message().c_str());
        }
        accept_handler(error);

    }

    void secure_session_impl::_receive()
    {
        ssl_socket_.async_read_some(
            boost::asio::buffer(recv_buffer_ + recv_buffer_used_ ,max_rcv_buf_size - recv_buffer_used_),
            boost::bind(&secure_session_impl::_receive_handler, shared_from_this(),
            boost::asio::placeholders::error,
            boost::asio::placeholders::bytes_transferred)
            );
    }

    void secure_session_impl::_send()
    {
        boost::asio::async_write(ssl_socket_, send_buffer_,
            boost::bind(&secure_session_impl::_send_handler, shared_from_this(), boost::asio::placeholders::error,
            boost::asio::placeholders::bytes_transferred));
    }
}   // namespace of network
}   // namespace of utility

